from app.utils.kafka import KafkaProducer
from app.utils.logger import logger


class MessageReceiver:
    def __init__(self):
        self.kafka_producer = KafkaProducer()
        self._started = False

    async def start(self):
        """启动 Kafka 生产者"""
        if not self._started:
            await self.kafka_producer.start()
            logger.info("Kafka producer started")
            self._started = True

    async def stop(self):
        """关闭 Kafka 生产者"""
        if self._started:
            await self.kafka_producer.stop()
            logger.info("Kafka producer stopped")
            self._started = False

    async def process(self, data):
        """处理消息并发送到 Kafka"""
        if not self._started:
            raise RuntimeError("Kafka producer is not started")
        await self.kafka_producer.send("qq_messages", data)
        logger.info(f"receiver passing data:{data}")
